package com.siteseat.crudeq.broker.v1.request;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureProgressListener;
import org.jboss.netty.channel.DefaultFileRegion;
import org.jboss.netty.channel.FileRegion;

import com.siteseat.common.security.SecurityException;
import com.siteseat.crudeq.broker.security.BrokerPermission;
import com.siteseat.crudeq.broker.transport.BrokerSession;
import com.siteseat.crudeq.broker.transport.QueueNotExistsException;
import com.siteseat.crudeq.common.raw.protocol.v1.V1ResponseId;
import com.siteseat.crudeq.kernel.MessageHistory;
import com.siteseat.crudeq.kernel.MessageId;
import com.siteseat.crudeq.kernel.exception.InvalidChunkException;

/**
 *
 */
public class V1HistoryHandler extends V1AbstractRequestHandler {

  private int sendFileSizeThreshold = 1024;
  
  /**
   * Gets the sendFileSizeThreshold
   *
   * @return the sendFileSizeThreshold
   */
  public int getSendFileSizeThreshold() {
    return sendFileSizeThreshold;
  }

  /**
   * Sets the sendFileSizeThreshold
   *
   * @param sendFileSizeThreshold the sendFileSizeThreshold to set
   */
  public void setSendFileSizeThreshold( int sendFileSizeThreshold ) {
    this.sendFileSizeThreshold = sendFileSizeThreshold;
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void execute( BrokerSession session, byte reqId, Channel ch, ChannelBuffer input, short version )
          throws Exception {
    Collection<MessageId> mids = parse( session, ch, input );
    if( mids == null || mids.size() == 0 ) {
      responseOK( ch, V1ResponseId.NO_MESSAGE );
      return;
    }

    List<MessageHistory> toSend = new ArrayList<MessageHistory>( mids.size() );
    int totalSize = fetch( session, mids, toSend );
    if( toSend.size() > 0 ) {
      send( toSend, totalSize, ch );
      return;
    } 

    responseOK( ch, V1ResponseId.NO_MESSAGE );
  }

  protected void subscribe() {

  }

  /**
   * @param toSend
   * @param totalSize
   * @param ch
   * @param buf
   * @throws FileNotFoundException 
   */
  protected void send( List<MessageHistory> toSend, int totalSize, Channel ch ) throws IOException {
    ChannelBuffer output = ChannelBuffers.dynamicBuffer();

    output.writeByte( V1ResponseId.MESSAGE_SET );// response id
    output.writeInt( totalSize );

    output.writeInt( toSend.size() );// queue count

    for( final MessageHistory mh: toSend ) {
      // queue id + chunk count
      output.writeInt( mh.getQueueId().intValue() );// queue id
      output.writeInt( mh.getSendInfos().size() );// chunk count

      // then use sendfile to send each chunk
      for( final MessageHistory.SendInfo si: mh.getSendInfos() ) {
        // (then)chunk id + chunk size
        output.writeLong( si.info.getChunkId().longValue() );// chunk id
        output.writeInt( si.offset );
        output.writeInt( si.size );// chunk size

        ch.write( output );
        output.clear();

        RandomAccessFile raf = new RandomAccessFile( si.info.getFile(), "r" );
        assert raf.length() == si.size;
        
        final FileRegion region = new DefaultFileRegion( raf.getChannel(), si.offset, si.size );
        ChannelFuture writeFuture = ch.write( region );
        writeFuture.addListener( new ChannelFutureProgressListener() {
          public void operationComplete( ChannelFuture future ) {
            region.releaseExternalResources();
          }

          public void operationProgressed( ChannelFuture future, long amount, long current, long total ) {
            // do nothing
          }
        } );
      }
    }
  }

  /**
   * @param session
   * @param mids
   * @param result
   * @return
   * @throws InvalidChunkException
   * @throws QueueNotExistsException
   */
  protected int fetch( BrokerSession session, Collection<MessageId> mids, List<MessageHistory> result )
          throws InvalidChunkException, QueueNotExistsException {

    long totalSize = 4 /* queue count */;

    for( MessageId mid: mids ) {
      MessageHistory mh = session.fetchMessage( mid );
      if( mh == null ) {
        continue;
      }
      int sendSize = mh.getTotalSendSize();
      if( totalSize + sendSize >= Integer.MAX_VALUE ) {
        continue;
      }
      totalSize += sendSize;
      result.add( mh );
    }

    return (int)totalSize;
  }

  /**
   * @param session
   * @param input
   * @return
   * @throws SecurityException
   */
  protected Collection<MessageId> parse( BrokerSession session, Channel ch, ChannelBuffer input )
          throws SecurityException {
    int msgCount = input.readInt();
    if( msgCount < 0 ) {
      writeAndLogInvalidRequestError( ch, session, "Message ID count must be > 0" );
      return null;
    }
    if( msgCount == 0 ) {
      return null;
    }

    Map<Integer, MessageId> mids = new HashMap<Integer, MessageId>( msgCount );

    for( int i = 0; i < msgCount; i++ ) {
      long chunkId = input.readLong();
      int offset = input.readInt();
      Integer queueId = Integer.valueOf( input.readInt() );

      session.ensureGranted( queueId.toString(), BrokerPermission.MESSAGE_SUBSCRIBE );

      MessageId mid = new MessageId( chunkId, offset, queueId );

      MessageId prev = mids.get( queueId );
      if( prev == null || mid.olderThan( prev ) ) {
        mids.put( queueId, mid );
      }
    }

    return mids.values();
  }

}
